package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    ////value:{"database":"gmall","table":"base_trademark","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,"data":{"id":"13","tm_name":"aaaa"},"old":null,"sinkTable":"dim_base_trademark"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        // 获取写入phoenix的表名和字段
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");

        // 生成sql
        String upsertSql = gensql(sinkTable, data);

        // 打印sql
        System.out.println(upsertSql);

        // 提交并执行sql
        PreparedStatement preparedStatement = connection.prepareStatement(upsertSql);

        if ("update".equals(value.getString("type"))){
            DimUtil.deleteRedisDimInfo(sinkTable.toUpperCase(), data.getString("id"));
        }

        preparedStatement.execute();
        connection.commit();

        // 关闭perparedStatement
        preparedStatement.close();

    }

    private String gensql(String sinkTable, JSONObject data) {
        return "upsert into " +
                GmallConfig.HBASE_SCHEMA +
                "." +
                sinkTable +
                "(" +
                StringUtils.join(data.keySet(), ",") +
                ") values('" +
                StringUtils.join(data.values(), "','") +
                "')";
    }
}
